package com.akkainaction.cluster.wordcount.front;

import akka.actor.ActorRef;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import com.akkainaction.cluster.wordcount.po.Article;
import java.util.ArrayList;
import java.util.List;

public class WordFrontService extends UntypedActor {

    private int jobCounter = 0;

    private List<ActorRef> wordCountServices = new ArrayList<>();

    @Override
    public void onReceive(Object msg) {
        if (msg instanceof Article) {
            jobCounter++;
            Article art = (Article) msg;
            // 通过取余选择后端节点
            int serviceNodeIndex = jobCounter % wordCountServices.size();
            System.out.println("选择节点:" + serviceNodeIndex);
            wordCountServices.get(serviceNodeIndex).forward(art, getContext());
        } else if (msg instanceof String) {
            String cmd = (String) msg;
            if (cmd.equals("serviceIsOK")) {
                // 后端Service已就绪
                ActorRef backendSender = getSender();
                System.out.println(backendSender + "可用");
                wordCountServices.add(backendSender);
                getContext().watch(backendSender);
            } else if (cmd.equals("isready")) {
                if (!wordCountServices.isEmpty()) {
                    getSender().tell("ready", getSelf());
                } else {
                    getSender().tell("notready", getSelf());
                }
            }
        } else if (msg instanceof Terminated) {
            // 当后端Service终止时，将其从后端列表中移除
            Terminated ter = (Terminated) msg;
            System.out.println("移除了" + ter.getActor());
            wordCountServices.remove(ter.getActor());
        } else {
            unhandled(msg);
        }
    }
}
